/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.atlas.hive.bridge;

import com.sun.jersey.api.client.ClientResponse;
import org.apache.atlas.ApplicationProperties;
import org.apache.atlas.AtlasClientV2;
import org.apache.atlas.AtlasException;
import org.apache.atlas.AtlasServiceException;
import org.apache.atlas.hive.hook.events.BaseHiveEvent;
import org.apache.atlas.hive.model.AuditRule;
import org.apache.atlas.hive.model.HiveDataTypes;
import org.apache.atlas.hook.AtlasHookException;
import org.apache.atlas.model.instance.*;
import org.apache.atlas.model.instance.AtlasEntity.AtlasEntityWithExtInfo;
import org.apache.atlas.model.typedef.AtlasEntityDef;
import org.apache.atlas.model.typedef.AtlasTypesDef;
import org.apache.atlas.type.AtlasTypeUtil;
import org.apache.atlas.utils.AuthenticationUtil;
import org.apache.atlas.utils.HdfsNameServiceResolver;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang.RandomStringUtils;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.api.*;
import org.apache.hadoop.hive.ql.metadata.Hive;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.hadoop.hive.ql.metadata.Table;
import static org.apache.atlas.hive.bridge.RuleDefOperation.RULE_PURPOSE;
import static org.apache.atlas.hive.bridge.RuleDefOperation.RULE_TASK;
import static org.apache.atlas.hive.hook.events.BaseHiveEvent.*;

/**
 * Created by sqy on 2018/12/14.
 */
public class RuleEntityUtil {
    private static final Logger LOG = LoggerFactory.getLogger(RuleEntityUtil.class);

    public static final String HIVE_CLUSTER_NAME               = "atlas.cluster.name";
    public static final String DEFAULT_CLUSTER_NAME            = "primary";
    public static final String TEMP_TABLE_PREFIX               = "_temp-";
    public static final String ATLAS_ENDPOINT                  = "atlas.rest.address";
    public static final String SEP                             = ":".intern();
    private static final String DEFAULT_ATLAS_URL = "http://localhost:21000/";

    private final String   clusterName;
    private final AtlasClientV2 atlasClientV2;
    private final Hive hiveClient;

    private  final static List<AuditRule> ruleList = JsonUtils.createRule();

    public RuleEntityUtil(String clusterName, AtlasClientV2 atlasClientV2, Hive hiveClient){
        this.clusterName = clusterName;
        this.atlasClientV2 = atlasClientV2;
        this.hiveClient = hiveClient;
    }

    public String getClusterName() {
        return clusterName;
    }

    public Hive getHiveClient() {
        return hiveClient;
    }

    /*public void init() {
        Configuration atlasConf = null;
        try {
            atlasConf = ApplicationProperties.get();
            String[] atlasEndpoint = atlasConf.getStringArray(ATLAS_ENDPOINT);
            if (atlasEndpoint == null || atlasEndpoint.length == 0) {
                atlasEndpoint = new String[] { DEFAULT_ATLAS_URL };
            }

            final AtlasClientV2 atlasClientV2;

            if (!AuthenticationUtil.isKerberosAuthenticationEnabled()) {
                String[] basicAuthUsernamePassword = AuthenticationUtil.getBasicAuthenticationInput();

                atlasClientV2 = new AtlasClientV2(atlasEndpoint, basicAuthUsernamePassword);
            } else {
                UserGroupInformation ugi = UserGroupInformation.getCurrentUser();

                atlasClientV2 = new AtlasClientV2(ugi, ugi.getShortUserName(), atlasEndpoint);
            }

            final String  clusterName = atlasConf.getString(HIVE_CLUSTER_NAME, DEFAULT_CLUSTER_NAME);
            RuleEntityUtil ruleEntityUtil = new RuleEntityUtil(clusterName,atlasClientV2, Hive.get(new HiveConf()));
            ruleEntityUtil.importAudit();

        } catch (AtlasException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }*/

    public void importAudit() throws Exception {

        AtlasEntityWithExtInfo purposeEntity;
        AtlasEntityWithExtInfo taskEntity;

        if (CollectionUtils.isNotEmpty(ruleList)){
            for (AuditRule rule : ruleList) {
                purposeEntity = registerPurpose(rule);
                LOG.info("sqy==>purposeEntity="+purposeEntity);

                if (CollectionUtils.isNotEmpty(rule.getTask())){
                    for (String task : rule.getTask()) {
                        taskEntity = registerTask(task,rule);
                        LOG.info("sqy==>taskEntity="+taskEntity);
                    }
                }
            }
        }
    }

    public AtlasEntityWithExtInfo registerPurpose(AuditRule rule) throws Exception {
        AtlasEntityWithExtInfo ret = null;
        ret = findPurpose(rule.getPurpose());

        if (ret == null){
            ret = registerInstance(new AtlasEntityWithExtInfo(toPurposeEntity(rule)));
        }else {

            ret.setEntity(toPurposeEntity(rule,ret.getEntity()));
            updateInstance(ret);
        }

        return ret;
    }

    public AtlasEntityWithExtInfo registerTask(String task, AuditRule rule) throws Exception {
        AtlasEntityWithExtInfo ret = null;
        List<AtlasEntity> inputs = new ArrayList<>();
        List<AtlasEntity> outputs = new ArrayList<>();

        AtlasEntityWithExtInfo input = findEntity(RULE_PURPOSE,rule.getPurpose());
        inputs.add(input.getEntity());

        Set<String> usedDataList = rule.getUsedData();

        if (CollectionUtils.isNotEmpty(usedDataList)){
            for (String data : usedDataList) {

                AtlasEntity entity = getOutPutEntity(data);
                outputs.add(entity);
            }
        }


        ret = findTask(task);
        if (ret == null){
            ret = registerInstance(new AtlasEntityWithExtInfo(toTaskEntity(task,rule,inputs,outputs)));
        }else {

            ret.setEntity(toTaskEntity(task,rule,inputs,outputs,ret.getEntity()));
            updateInstance(ret);
        }
        return ret;
    }

    private AtlasEntity getOutPutEntity(String data) throws Exception {

        Pattern p = Pattern.compile("[.]");
        Matcher m = p.matcher(data);
        boolean flag = m.find();

        String[] strings = data.split("\\.");

        if (!flag){//db

            AtlasEntityWithExtInfo ret = registerDatabase(data);
            return ret.getEntity();

        }else {//table
            String dbName = strings[0];
            String tableName = strings[1];

            AtlasEntityWithExtInfo dbret = registerDatabase(dbName);

            Table table = hiveClient.getTable(dbName,tableName);
            AtlasEntityWithExtInfo tableret = registerTable(dbret.getEntity(),table);
            return tableret.getEntity();
        }

    }

    private AtlasEntityWithExtInfo registerDatabase(String databaseName) throws Exception {
        AtlasEntityWithExtInfo ret = null;
        Database               db  = hiveClient.getDatabase(databaseName);

        if (db != null) {
            ret = findDatabase(clusterName, databaseName);

            if (ret == null) {
                ret = registerInstance(new AtlasEntityWithExtInfo(toDbEntity(db)));
            } else {
                LOG.info("Database {} is already registered - id={}. Updating it.", databaseName, ret.getEntity().getGuid());

                ret.setEntity(toDbEntity(db, ret.getEntity()));

                updateInstance(ret);
            }
        }

        return ret;
    }

    private AtlasEntity toDbEntity(Database hiveDB) throws HiveException {
        return toDbEntity(hiveDB, null);
    }

    private AtlasEntity toDbEntity(Database hiveDB, AtlasEntity dbEntity) {
        if (dbEntity == null) {
            dbEntity = new AtlasEntity(HiveDataTypes.HIVE_DB.getName());
        }

        String dbName = hiveDB.getName().toLowerCase();

        dbEntity.setAttribute(ATTRIBUTE_QUALIFIED_NAME, getDBQualifiedName(clusterName, dbName));
        dbEntity.setAttribute(ATTRIBUTE_NAME, dbName);
        dbEntity.setAttribute(ATTRIBUTE_DESCRIPTION, hiveDB.getDescription());
        dbEntity.setAttribute(ATTRIBUTE_OWNER, hiveDB.getOwnerName());

        dbEntity.setAttribute(ATTRIBUTE_CLUSTER_NAME, clusterName);
        dbEntity.setAttribute(ATTRIBUTE_LOCATION, HdfsNameServiceResolver.getPathWithNameServiceID(hiveDB.getLocationUri()));
        dbEntity.setAttribute(ATTRIBUTE_PARAMETERS, hiveDB.getParameters());

        if (hiveDB.getOwnerType() != null) {
            dbEntity.setAttribute(ATTRIBUTE_OWNER_TYPE, OWNER_TYPE_TO_ENUM_VALUE.get(hiveDB.getOwnerType().getValue()));
        }

        return dbEntity;
    }

    private AtlasEntityWithExtInfo findDatabase(String clusterName, String databaseName) throws AtlasServiceException {
        String typeName = HiveDataTypes.HIVE_DB.getName();

        return findEntity(typeName, getDBQualifiedName(clusterName, databaseName));
    }

    private AtlasEntityWithExtInfo registerTable(AtlasEntity dbEntity, Table table) throws AtlasHookException {
        try {
            AtlasEntityWithExtInfo ret;
            AtlasEntityWithExtInfo tableEntity = findTableEntity(table);

            if (tableEntity == null) {
                tableEntity = toTableEntity(dbEntity, table);

                ret = registerInstance(tableEntity);
            } else {
                LOG.info("Table {}.{} is already registered with id {}. Updating entity.", table.getDbName(), table.getTableName(), tableEntity.getEntity().getGuid());

                ret = toTableEntity(dbEntity, table, tableEntity);

                updateInstance(ret);
            }

            return ret;
        } catch (Exception e) {
            throw new AtlasHookException("HiveMetaStoreBridge.registerTable() failed.", e);
        }
    }

    private AtlasEntityWithExtInfo findTableEntity(Table hiveTable)  throws Exception {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Searching Atlas for table {}.{}", hiveTable.getDbName(), hiveTable.getTableName());
        }

        String typeName         = HiveDataTypes.HIVE_TABLE.getName();
        String tblQualifiedName = getTableQualifiedName(getClusterName(), hiveTable.getDbName(), hiveTable.getTableName());

        return findEntity(typeName, tblQualifiedName);
    }

    private AtlasEntityWithExtInfo toTableEntity(AtlasEntity database, Table hiveTable) throws AtlasHookException {
        return toTableEntity(database, hiveTable, null);
    }

    private AtlasEntityWithExtInfo toTableEntity(AtlasEntity database, final Table hiveTable, AtlasEntityWithExtInfo table) throws AtlasHookException {
        if (table == null) {
            table = new AtlasEntityWithExtInfo(new AtlasEntity(HiveDataTypes.HIVE_TABLE.getName()));
        }

        AtlasEntity tableEntity        = table.getEntity();
        String      tableQualifiedName = getTableQualifiedName(clusterName, hiveTable);
        long        createTime         = BaseHiveEvent.getTableCreateTime(hiveTable);
        long        lastAccessTime     = hiveTable.getLastAccessTime() > 0 ? hiveTable.getLastAccessTime() : createTime;

        tableEntity.setAttribute(ATTRIBUTE_DB, BaseHiveEvent.getObjectId(database));
        tableEntity.setAttribute(ATTRIBUTE_QUALIFIED_NAME, tableQualifiedName);
        tableEntity.setAttribute(ATTRIBUTE_NAME, hiveTable.getTableName().toLowerCase());
        tableEntity.setAttribute(ATTRIBUTE_OWNER, hiveTable.getOwner());

        tableEntity.setAttribute(ATTRIBUTE_CREATE_TIME, createTime);
        tableEntity.setAttribute(ATTRIBUTE_LAST_ACCESS_TIME, lastAccessTime);
        tableEntity.setAttribute(ATTRIBUTE_RETENTION, hiveTable.getRetention());
        tableEntity.setAttribute(ATTRIBUTE_PARAMETERS, hiveTable.getParameters());
        tableEntity.setAttribute(ATTRIBUTE_COMMENT, hiveTable.getParameters().get(ATTRIBUTE_COMMENT));
        tableEntity.setAttribute(ATTRIBUTE_TABLE_TYPE, hiveTable.getTableType().name());
        tableEntity.setAttribute(ATTRIBUTE_TEMPORARY, hiveTable.isTemporary());

        if (hiveTable.getViewOriginalText() != null) {
            tableEntity.setAttribute(ATTRIBUTE_VIEW_ORIGINAL_TEXT, hiveTable.getViewOriginalText());
        }

        if (hiveTable.getViewExpandedText() != null) {
            tableEntity.setAttribute(ATTRIBUTE_VIEW_EXPANDED_TEXT, hiveTable.getViewExpandedText());
        }

        AtlasEntity       sdEntity = toStroageDescEntity(hiveTable.getSd(), tableQualifiedName, getStorageDescQFName(tableQualifiedName), BaseHiveEvent.getObjectId(tableEntity));
        List<AtlasEntity> partKeys = toColumns(hiveTable.getPartitionKeys(), tableEntity);
        List<AtlasEntity> columns  = toColumns(hiveTable.getCols(), tableEntity);

        tableEntity.setAttribute(ATTRIBUTE_STORAGEDESC, BaseHiveEvent.getObjectId(sdEntity));
        tableEntity.setAttribute(ATTRIBUTE_PARTITION_KEYS, BaseHiveEvent.getObjectIds(partKeys));
        tableEntity.setAttribute(ATTRIBUTE_COLUMNS, BaseHiveEvent.getObjectIds(columns));

        table.addReferredEntity(database);
        table.addReferredEntity(sdEntity);

        if (partKeys != null) {
            for (AtlasEntity partKey : partKeys) {
                table.addReferredEntity(partKey);
            }
        }

        if (columns != null) {
            for (AtlasEntity column : columns) {
                table.addReferredEntity(column);
            }
        }

        table.setEntity(tableEntity);

        return table;
    }

    private AtlasEntity toStroageDescEntity(StorageDescriptor storageDesc, String tableQualifiedName, String sdQualifiedName, AtlasObjectId tableId ) throws AtlasHookException {
        AtlasEntity ret = new AtlasEntity(HiveDataTypes.HIVE_STORAGEDESC.getName());

        ret.setAttribute(ATTRIBUTE_TABLE, tableId);
        ret.setAttribute(ATTRIBUTE_QUALIFIED_NAME, sdQualifiedName);
        ret.setAttribute(ATTRIBUTE_PARAMETERS, storageDesc.getParameters());
        ret.setAttribute(ATTRIBUTE_LOCATION, HdfsNameServiceResolver.getPathWithNameServiceID(storageDesc.getLocation()));
        ret.setAttribute(ATTRIBUTE_INPUT_FORMAT, storageDesc.getInputFormat());
        ret.setAttribute(ATTRIBUTE_OUTPUT_FORMAT, storageDesc.getOutputFormat());
        ret.setAttribute(ATTRIBUTE_COMPRESSED, storageDesc.isCompressed());
        ret.setAttribute(ATTRIBUTE_NUM_BUCKETS, storageDesc.getNumBuckets());
        ret.setAttribute(ATTRIBUTE_STORED_AS_SUB_DIRECTORIES, storageDesc.isStoredAsSubDirectories());

        if (storageDesc.getBucketCols().size() > 0) {
            ret.setAttribute(ATTRIBUTE_BUCKET_COLS, storageDesc.getBucketCols());
        }

        if (storageDesc.getSerdeInfo() != null) {
            SerDeInfo serdeInfo = storageDesc.getSerdeInfo();

            LOG.debug("serdeInfo = {}", serdeInfo);
            // SkewedInfo skewedInfo = storageDesc.getSkewedInfo();

            AtlasStruct serdeInfoStruct = new AtlasStruct(HiveDataTypes.HIVE_SERDE.getName());

            serdeInfoStruct.setAttribute(ATTRIBUTE_NAME, serdeInfo.getName());
            serdeInfoStruct.setAttribute(ATTRIBUTE_SERIALIZATION_LIB, serdeInfo.getSerializationLib());
            serdeInfoStruct.setAttribute(ATTRIBUTE_PARAMETERS, serdeInfo.getParameters());

            ret.setAttribute(ATTRIBUTE_SERDE_INFO, serdeInfoStruct);
        }

        if (CollectionUtils.isNotEmpty(storageDesc.getSortCols())) {
            List<AtlasStruct> sortColsStruct = new ArrayList<>();

            for (Order sortcol : storageDesc.getSortCols()) {
                String hiveOrderName = HiveDataTypes.HIVE_ORDER.getName();
                AtlasStruct colStruct = new AtlasStruct(hiveOrderName);
                colStruct.setAttribute("col", sortcol.getCol());
                colStruct.setAttribute("order", sortcol.getOrder());

                sortColsStruct.add(colStruct);
            }

            ret.setAttribute(ATTRIBUTE_SORT_COLS, sortColsStruct);
        }

        return ret;
    }

    private List<AtlasEntity> toColumns(List<FieldSchema> schemaList, AtlasEntity table) throws AtlasHookException {
        List<AtlasEntity> ret = new ArrayList<>();

        int columnPosition = 0;
        for (FieldSchema fs : schemaList) {
            LOG.debug("Processing field {}", fs);

            AtlasEntity column = new AtlasEntity(HiveDataTypes.HIVE_COLUMN.getName());

            column.setAttribute(ATTRIBUTE_TABLE, BaseHiveEvent.getObjectId(table));
            column.setAttribute(ATTRIBUTE_QUALIFIED_NAME, getColumnQualifiedName((String) table.getAttribute(ATTRIBUTE_QUALIFIED_NAME), fs.getName()));
            column.setAttribute(ATTRIBUTE_NAME, fs.getName());
            column.setAttribute(ATTRIBUTE_OWNER, table.getAttribute(ATTRIBUTE_OWNER));
            column.setAttribute(ATTRIBUTE_COL_TYPE, fs.getType());
            column.setAttribute(ATTRIBUTE_COL_POSITION, columnPosition++);
            column.setAttribute(ATTRIBUTE_COMMENT, fs.getComment());

            ret.add(column);
        }
        return ret;
    }

    public static String getStorageDescQFName(String tableQualifiedName) {
        return tableQualifiedName + "_storage";
    }

    public static String getColumnQualifiedName(final String tableQualifiedName, final String colName) {
        final String[] parts       = tableQualifiedName.split("@");
        final String   tableName   = parts[0];
        final String   clusterName = parts[1];

        return String.format("%s.%s@%s", tableName, colName.toLowerCase(), clusterName);
    }

    public static String getTableQualifiedName(String clusterName, String dbName, String tableName, boolean isTemporaryTable) {
        String tableTempName = tableName;

        if (isTemporaryTable) {
            if (SessionState.get() != null && SessionState.get().getSessionId() != null) {
                tableTempName = tableName + TEMP_TABLE_PREFIX + SessionState.get().getSessionId();
            } else {
                tableTempName = tableName + TEMP_TABLE_PREFIX + RandomStringUtils.random(10);
            }
        }

        return String.format("%s.%s@%s", dbName.toLowerCase(), tableTempName.toLowerCase(), clusterName);
    }

    public static String getTableQualifiedName(String clusterName, String dbName, String tableName) {
        return getTableQualifiedName(clusterName, dbName, tableName, false);
    }

    private static String getTableQualifiedName(String clusterName, Table table) {
        return getTableQualifiedName(clusterName, table.getDbName(), table.getTableName(), table.isTemporary());
    }
    /*public AtlasEntityWithExtInfo registerUsedData(AuditRule rule, List<AtlasEntity> inputs, List<AtlasEntity> outputs) throws Exception {
        AtlasEntityWithExtInfo ret = null;

        Set<String> usedDatas = rule.getUsedData();
        if (CollectionUtils.isNotEmpty(usedDatas)){
            for (String usedData : usedDatas) {

                String[] strings = usedData.split("\\.");
                String dbName = strings[0];
                String tableName = strings[1];
                String table = (dbName + QNAME_SEP_ENTITY_NAME + tableName + QNAME_SEP_CLUSTER_NAME).toLowerCase() + getClusterName();

                ret = findUsedData(table);
                if (ret == null){
                    ret = registerInstance(new AtlasEntity.AtlasEntityWithExtInfo(toUsedData(usedData,rule,inputs,outputs)));
                }else {
                    ret.setEntity(toUsedDataEntity(usedData,rule,inputs,outputs,ret.getEntity()));
                    updateInstance(ret);
                }
            }
        }

        return ret;
    }*/


    public  AtlasEntity toPurposeEntity(AuditRule rule) {
        return toPurposeEntity(rule,null);
    }

    public AtlasEntity toPurposeEntity(AuditRule rule, AtlasEntity entity) {
        if(entity == null){
            entity = new AtlasEntity(RULE_PURPOSE);
        }

        entity.setAttribute(ATTRIBUTE_NAME,rule.getPurpose());
        entity.setAttribute(ATTRIBUTE_QUALIFIED_NAME,rule.getPurpose());

        entity.setAttribute(ATTRIBUTE_DESCRIPTION,rule.getDescription());

        return entity;
    }

    public  AtlasEntity toTaskEntity(String task, AuditRule rule, List<AtlasEntity> inputs,List<AtlasEntity> outputs) {
        return toTaskEntity(task,rule,inputs,outputs,null);
    }

    public AtlasEntity toTaskEntity(String task,AuditRule rule,List<AtlasEntity> inputs,List<AtlasEntity> outputs, AtlasEntity entity) {
        if(entity == null){
            entity = new AtlasEntity(RULE_TASK);
        }
        entity.setAttribute(ATTRIBUTE_NAME,task);
        entity.setAttribute(ATTRIBUTE_QUALIFIED_NAME,task);
        entity.setAttribute(ATTRIBUTE_INPUTS, BaseHiveEvent.getObjectIds(inputs));
        entity.setAttribute(ATTRIBUTE_OUTPUTS,BaseHiveEvent.getObjectIds(outputs));
        entity.setAttribute(ATTRIBUTE_DESCRIPTION,rule.getDescription());

        return entity;

    }

    /*public AtlasEntity toUsedDataEntity(String usedData, AuditRule rule, List<AtlasEntity> inputs,List<AtlasEntity> outputs,AtlasEntity entity) {
        if(entity == null){
            entity = new AtlasEntity(HiveDataTypes.HIVE_TABLE.getName());
        }

        String[] strings = usedData.split("\\.");
        String dbName = strings[0];
        String tableName = strings[1];
        String qualifitedName = (dbName + QNAME_SEP_ENTITY_NAME + tableName + QNAME_SEP_CLUSTER_NAME).toLowerCase() + getClusterName();

        entity.setAttribute(ATTRIBUTE_NAME,tableName);
        entity.setAttribute(ATTRIBUTE_QUALIFIED_NAME,qualifitedName);

        return entity;
    }

    public AtlasEntity toUsedData(String table, AuditRule rule,List<AtlasEntity> inputs,List<AtlasEntity> outputs) {
        return toUsedDataEntity(table,rule,inputs,outputs,null);
    }*/

    public AtlasEntityWithExtInfo findPurpose(String purpose) throws AtlasServiceException {
        String typeName = RULE_PURPOSE;
        return findEntity(typeName,purpose);
    }

    public AtlasEntityWithExtInfo findTask(String task) throws AtlasServiceException {
        String typeName = RULE_TASK;
        return findEntity(typeName,task);
    }

   /* public AtlasEntityWithExtInfo findUsedData(String usedData) throws AtlasServiceException {
        String typeName = HiveDataTypes.HIVE_TABLE.getName();
        return findEntity(typeName,usedData);
    }*/

    private AtlasEntityWithExtInfo registerInstance(AtlasEntityWithExtInfo entity) throws Exception {
        if (LOG.isDebugEnabled()) {
            LOG.debug("creating {} entity: {}", entity.getEntity().getTypeName(), entity);
        }

        AtlasEntityWithExtInfo  ret             = null;
        EntityMutationResponse  response        = atlasClientV2.createEntity(entity);
        List<AtlasEntityHeader> createdEntities = response.getEntitiesByOperation(EntityMutations.EntityOperation.CREATE);

        if (CollectionUtils.isNotEmpty(createdEntities)) {
            for (AtlasEntityHeader createdEntity : createdEntities) {
                if (ret == null) {
                    ret = atlasClientV2.getEntityByGuid(createdEntity.getGuid());

                    LOG.info("Created {} entity: name={}, guid={}", ret.getEntity().getTypeName(), ret.getEntity().getAttribute(ATTRIBUTE_QUALIFIED_NAME), ret.getEntity().getGuid());
                } else if (ret.getEntity(createdEntity.getGuid()) == null) {
                    AtlasEntityWithExtInfo newEntity = atlasClientV2.getEntityByGuid(createdEntity.getGuid());

                    ret.addReferredEntity(newEntity.getEntity());

                    if (MapUtils.isNotEmpty(newEntity.getReferredEntities())) {
                        for (Map.Entry<String, AtlasEntity> entry : newEntity.getReferredEntities().entrySet()) {
                            ret.addReferredEntity(entry.getKey(), entry.getValue());
                        }
                    }

                    LOG.info("Created {} entity: name={}, guid={}", newEntity.getEntity().getTypeName(), newEntity.getEntity().getAttribute(ATTRIBUTE_QUALIFIED_NAME), newEntity.getEntity().getGuid());
                }
            }
        }

        clearRelationshipAttributes(ret);

        return ret;
    }

    private void updateInstance(AtlasEntityWithExtInfo entity) throws AtlasServiceException {
        if (LOG.isDebugEnabled()) {
            LOG.debug("updating {} entity: {}", entity.getEntity().getTypeName(), entity);
        }

        atlasClientV2.updateEntity(entity);

        LOG.info("Updated {} entity: name={}, guid={}", entity.getEntity().getTypeName(), entity.getEntity().getAttribute(ATTRIBUTE_QUALIFIED_NAME), entity.getEntity().getGuid());
    }

    private AtlasEntityWithExtInfo findEntity(final String typeName, final String qualifiedName) throws AtlasServiceException {
        AtlasEntityWithExtInfo ret = null;

        try {
            ret = atlasClientV2.getEntityByAttribute(typeName, Collections.singletonMap(ATTRIBUTE_QUALIFIED_NAME, qualifiedName));
        } catch (AtlasServiceException e) {
            if(e.getStatus() == ClientResponse.Status.NOT_FOUND) {
                return null;
            }

            throw e;
        }

        clearRelationshipAttributes(ret);

        return ret;
    }

    public static String getDBQualifiedName(String clusterName, String dbName) {
        return String.format("%s@%s", dbName.toLowerCase(), clusterName);
    }

    private void clearRelationshipAttributes(AtlasEntityWithExtInfo entity) {
        if (entity != null) {
            clearRelationshipAttributes(entity.getEntity());

            if (entity.getReferredEntities() != null) {
                clearRelationshipAttributes(entity.getReferredEntities().values());
            }
        }
    }

    private void clearRelationshipAttributes(Collection<AtlasEntity> entities) {
        if (entities != null) {
            for (AtlasEntity entity : entities) {
                clearRelationshipAttributes(entity);
            }
        }
    }

    private void clearRelationshipAttributes(AtlasEntity entity) {
        if (entity != null && entity.getRelationshipAttributes() != null) {
            entity.getRelationshipAttributes().clear();
        }
    }

    public  void createTypes() throws AtlasServiceException {
        RuleDefOperation rulrDef = new RuleDefOperation();

        AtlasTypesDef atlasTypesDef = rulrDef.createRuleTypeDef();

        atlasClientV2.createAtlasTypeDefs(atlasTypesDef);
    }

    public void deleteType() throws AtlasServiceException {
        AtlasEntityDef purposeDef = atlasClientV2.getEntityDefByName("purpose");
        AtlasEntityDef usedDataDef = atlasClientV2.getEntityDefByName("usedData");
        AtlasEntityDef taskDef = atlasClientV2.getEntityDefByName("task");

        AtlasTypesDef atlasTypesDef = AtlasTypeUtil.getTypesDef(Collections.emptyList(),Collections.emptyList(),Collections.emptyList(),Arrays.asList(purposeDef,usedDataDef,taskDef));

        atlasClientV2.deleteAtlasTypeDefs(atlasTypesDef);
    }

     /*  public void deleteUsedData() throws AtlasServiceException {

        List<String> usedDataList = new ArrayList<>();
        usedDataList.add("booklendinfo");
        usedDataList.add("internetinfo");
        usedDataList.add("schoolinfo");
        usedDataList.add("financeinfo");

        String typeName = HiveDataTypes.HIVE_TABLE.getName();

        List<String> guids = new ArrayList<>();
        for (String usedData : usedDataList) {
            AtlasEntityWithExtInfo ret = findEntity(typeName,usedData);
            LOG.info("sqy==>ret="+ret);
            if (ret != null){
                guids.add(ret.getEntity().getGuid());
            }
        }
        LOG.info("sqy==>guids="+guids);
        atlasClientV2.deleteEntitiesByGuids(guids);
    }

    public void deletePurpose() throws AtlasServiceException {
        List<String> purposeList = new ArrayList<>();
        purposeList.add("internetAndbooklendAnalyse");
        purposeList.add("internetAnalyse");
        purposeList.add("SchoolAnalyse");
        purposeList.add("financeAnalyse");

        String typeName = PURPOSE_TYPE;
        List<String> guids = new ArrayList<>();

        for (String purpose : purposeList) {
            AtlasEntityWithExtInfo ret = findEntity(typeName,purpose);
            if (ret != null){
                guids.add(ret.getEntity().getGuid());
            }else {
                return;
            }
        }

        atlasClientV2.deleteEntitiesByGuids(guids);
    }

    public void deleteTask() throws AtlasServiceException {
        List<String> taskList = new ArrayList<>();
        taskList.add("avg");
        taskList.add("dense_rank");
        taskList.add("count");
        taskList.add("sum");
        taskList.add("rank");
        taskList.add("min");
        taskList.add("max");

        String typeName = TASK_TYPE;
        List<String> guids = new ArrayList<>();

        for (String task : taskList) {
            AtlasEntityWithExtInfo ret = findEntity(typeName,task);
            guids.add(ret.getEntity().getGuid());
        }

        atlasClientV2.deleteEntitiesByGuids(guids);
    }*/

}
